package FlinkSQL

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Kafka, Schema}


/**
 *
 * @program: FlinkSQL
 * @author: YCLW058
 * @create: 2021-05-25 17:37
 * @decsription:
 *
 * */

object FileOutputTest {

  def main(args: Array[String]): Unit = {
    //1 env
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tableEnv = StreamTableEnvironment.create(env)
    //2 source
    //2.1 从文件 读取数据
    val inputPath="data/input/test"
    tableEnv.connect(new FileSystem().path(inputPath))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id",DataTypes.INT())
        .field("name",DataTypes.STRING())
        .field("age",DataTypes.INT())
      ).createTemporaryTable("inputTable")

//    //2.2 从 kafka读取数据
//    tableEnv.connect(new Kafka()
//      //版本
//      .version("universal")
//      //主题
//      .topic("hello")
//      //zookeeper
//      .property("zookeeper.connect", "localhost:2181")
//      .property("bootstrap.servers", "localhost:9092")
//    )
//      .withFormat(new Csv())
//      .withSchema(new Schema()
//        .field("id",DataTypes.INT())
//        .field("name",DataTypes.STRING())
//        .field("age",DataTypes.INT())
//      ).createTemporaryTable("kafkaInputTable")

    //注册成表
    val inputTable = tableEnv.from("inputTable")

    //3 transformation
    // 3.1 使用 table API
    val resultTable = inputTable
      .select("id,name,age")

    // 3.2 使用SQL
    val resultSQLTable = tableEnv.sqlQuery(
      """
        |select id,name,age
        |from inputTable
        |""".stripMargin
    )


    val SQLtable = tableEnv.sqlQuery(
      """
        |select id,name,age
        |from inputTable
        |""".stripMargin
    )





    //4 sink
  //aggTable.toRetractStream[(Int,BigInt)].print()
    SQLtable.toAppendStream[(Int,String,Int)].print()

    //5 execute
    env.execute()

  }

}

